package org.example.function;

import com.antgroup.geaflow.api.context.RuntimeContext;
import com.antgroup.geaflow.api.function.RichFunction;
import com.antgroup.geaflow.api.function.io.SinkFunction;
import org.example.csv.CSVData;
import org.example.util.CSVFaster2;
import org.example.util.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;

public class CSVFileSink<OUT> extends RichFunction implements SinkFunction<OUT> {

    private static final Logger LOG = LoggerFactory.getLogger(CSVFileSink.class);

    private Path path;

    private String resultName;

    public static final String TASK_ID = "task.id";

    public static final String OUTPUT_DIR = "output.dir";

    private CSVData csvData;

    public CSVFileSink() {

    }

    public CSVFileSink(String resultName) {
        this.resultName = resultName;
    }

    @Override
    public void open(RuntimeContext runtimeContext) {
        String outputDir = runtimeContext.getConfiguration().getString(OUTPUT_DIR);
        // String taskId = runtimeContext.getConfiguration().getString(TASK_ID);
        Path outputAbsolutePath = Paths.get(outputDir).toAbsolutePath();
        path = FileUtils.pathJoin(outputAbsolutePath.toString(), this.resultName);
        // FileUtils.createFileIfExists(path);
        csvData = new CSVData(Arrays.asList("id", "value"));
    }

    @Override
    public void close() {
        this.csvData.sortValues();
        // CSVUtils.write(path, this.csvData);
        // CSVFaster.write(path, this.csvData);
        CSVFaster2.write(path, this.csvData);
    }

    @Override
    public void write(OUT out) throws Exception {
        this.csvData.addLine((String[]) out);
    }
}
